package net.lesscoding.rpc.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;
import net.lesscoding.common.Const;
import net.lesscoding.protocol.ProtocolFrameDecoder;
import net.lesscoding.protocol.SequenceIdGenerator;
import net.lesscoding.protocol.SharedMessageCodec;
import net.lesscoding.rpc.handler.RpcResponseMessageHandler;
import net.lesscoding.rpc.message.RpcRequestMessage;
import net.lesscoding.rpc.service.HelloService;

import java.lang.reflect.Proxy;

/**
 * @author eleven
 * @date 2024/12/4 8:39
 * @apiNote
 */
@Slf4j
public class RpcClientManager {
    private static volatile Channel channel;
    private static final Object LOCK = new Object();

    /**
     * 测试方法
     * 当前存在一个问题，就是所有的方法都需要转换成这种反射的形式调用
     * 相较于传统的 new HelloService().sayHell("李四") 有些麻烦
     * 还需要进一步的改进
     *
     * @param args
     */
    public static void main(String[] args) {
        HelloService proxyService = getProxyService(HelloService.class);
        System.out.println("\n响应结果：" + proxyService.sayHello("李四"));
        System.out.println("\n响应结果：" + proxyService.sayHello("张三"));
    }

    /**
     * 创建代理对象
     *
     * @param serviceClass 代理对象的接口
     * @param <T>          泛型
     * @return 代理对象
     */
    public static <T> T getProxyService(Class<T> serviceClass) {
        ClassLoader loader = serviceClass.getClassLoader();
        Class<?>[] interfaces = new Class[]{serviceClass};
        // 代理对象任何调用的方法都会被后边的callback接收到
        Object o = Proxy.newProxyInstance(loader, interfaces, (proxy, method, args) -> {
            // 1. 将方法调用转换成消息对象
            RpcRequestMessage msg = new RpcRequestMessage(
                    SequenceIdGenerator.nextId(),
                    serviceClass.getName(),
                    method.getName(),
                    method.getReturnType(),
                    method.getParameterTypes(),
                    args
            );
            // 2. 将消息对象发出去
            channel().writeAndFlush(msg);
            // 3. 创建一个空的Promise对象来接收结果
            DefaultPromise<?> promise = new DefaultPromise<>(channel().eventLoop());
            RpcResponseMessageHandler.PROMISE_MAP.put(msg.getSequenceId(), promise);
            // 4. 等待await结果
            Promise<?> await = promise.await();
            log.info("{}", await);
            // 5. 结果返回
            if (await.isSuccess()) {
                return promise.getNow();
            } else {
                throw new RuntimeException(await.cause().getMessage());
            }
        });
        return (T) o;
    }

    public static Channel channel() {
        if (channel != null) {
            return channel;
        }
        /**
         * 防止有两个线程 t1 t2同时进来时 channel 为空。
         * t1 使用 initChannel 创建线程之后，t2进来再次初始化 channel
         * 也称之为双检锁, 在 1.5之前可能因为指令重排序出现问题，
         * 使用 volatile 防止指令重排序, 保证双检锁的正确性
         */
        synchronized (LOCK) {
            if (channel != null) {
                return channel;
            }
            initChannel();
            return channel;
        }
    }

    /**
     * 初始化channel
     */
    private static void initChannel() {
        LoggingHandler loggingHandler = new LoggingHandler();
        SharedMessageCodec messageCodec = new SharedMessageCodec();
        NioEventLoopGroup group = new NioEventLoopGroup();
        RpcResponseMessageHandler rpcResponseMessageHandler = new RpcResponseMessageHandler();
        Bootstrap bootstrap = new Bootstrap()
                .group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline()
                                .addLast(new ProtocolFrameDecoder())
                                .addLast(loggingHandler)
                                .addLast(messageCodec)
                                .addLast(rpcResponseMessageHandler);
                    }
                });
        try {
            channel = bootstrap.connect(Const.HOST, Const.PORT).sync().channel();
            // 关闭channel不能使用sync让客户端阻塞，所以使用addListener
            channel.closeFuture().addListener(future -> group.shutdownGracefully());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
